#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RedFire 数据一致性检查服务
========================

服务功能:
- 定时数据一致性检查
- 实时监控和告警
- 自动修复建议
- Web API管理接口

Author: RedFire Team
Date: 2025-09-16
Version: v1.0.0
"""

import asyncio
import logging
import os
import sys
import signal
import json
from pathlib import Path
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
import yaml
from dataclasses import asdict
import schedule
import uvicorn
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse, HTMLResponse
import psutil

# 添加项目路径
current_dir = Path(__file__).parent
project_root = current_dir.parent.parent
sys.path.insert(0, str(project_root))

from databases.consistency.data_consistency_checker import (
    DataConsistencyChecker,
    ConsistencyReport,
    InconsistencyRecord,
    SeverityLevel,
    quick_consistency_check
)

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class ConsistencyService:
    """数据一致性检查服务"""
    
    def __init__(self, config_path: str = None):
        self.config_path = config_path or os.path.join(current_dir, "consistency_config.yaml")
        self.config = None
        self.consistency_checker = None
        self.running = False
        self.scheduler_task = None
        
        # 服务统计
        self.stats = {
            'service_start_time': None,
            'total_checks_performed': 0,
            'last_check_time': None,
            'total_inconsistencies_found': 0,
            'critical_issues_found': 0,
            'last_consistency_scores': {},
            'errors': 0
        }
        
        # 检查历史
        self.check_history: List[Dict[str, Any]] = []
        self.max_history_size = 1000
    
    async def load_config(self) -> Dict[str, Any]:
        """加载配置文件"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                config_data = yaml.safe_load(f)
            
            # 环境变量替换
            config_data = self._replace_env_vars(config_data)
            
            # 获取环境特定配置
            env = os.getenv('REDFIRE_ENV', 'development')
            if env in config_data.get('environments', {}):
                env_config = config_data['environments'][env]
                config_data = self._merge_config(config_data, env_config)
            
            self.config = config_data
            logger.info(f"Consistency service configuration loaded (env: {env})")
            return config_data
            
        except Exception as e:
            logger.error(f"Failed to load config: {e}")
            raise
    
    def _replace_env_vars(self, data):
        """递归替换环境变量"""
        if isinstance(data, dict):
            return {k: self._replace_env_vars(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [self._replace_env_vars(item) for item in data]
        elif isinstance(data, str) and data.startswith('${') and data.endswith('}'):
            env_expr = data[2:-1]
            if ':-' in env_expr:
                var_name, default_value = env_expr.split(':-', 1)
                return os.getenv(var_name, default_value)
            else:
                return os.getenv(env_expr, data)
        else:
            return data
    
    def _merge_config(self, base_config: Dict, env_config: Dict) -> Dict:
        """合并环境特定配置"""
        result = base_config.copy()
        
        def deep_merge(target, source):
            for key, value in source.items():
                if key in target and isinstance(target[key], dict) and isinstance(value, dict):
                    deep_merge(target[key], value)
                else:
                    target[key] = value
        
        deep_merge(result, env_config)
        return result
    
    async def initialize(self):
        """初始化一致性检查服务"""
        await self.load_config()
        
        # 创建数据一致性检查器
        mysql_config = self.config['databases']['mysql']
        postgresql_config = self.config['databases']['postgresql']
        
        self.consistency_checker = DataConsistencyChecker(mysql_config, postgresql_config)
        await self.consistency_checker.initialize()
        
        logger.info("Consistency service initialized successfully")
    
    async def start_service(self):
        """启动一致性检查服务"""
        if self.running:
            logger.warning("Consistency service is already running")
            return
        
        self.running = True
        self.stats['service_start_time'] = datetime.now()
        
        logger.info("Starting data consistency checking service...")
        
        # 执行初始检查
        await self.perform_initial_check()
        
        # 启动定时任务调度器
        self.scheduler_task = asyncio.create_task(self._scheduler_loop())
        
        logger.info("Consistency service started successfully")
    
    async def perform_initial_check(self):
        """执行初始检查"""
        logger.info("Performing initial consistency check...")
        
        try:
            # 执行快速检查
            result = await self._run_check('quick_check')
            
            self.stats['total_checks_performed'] += 1
            self.stats['last_check_time'] = datetime.now()
            
            # 处理检查结果
            await self._process_check_result(result, 'initial_check')
            
            logger.info("Initial consistency check completed")
            
        except Exception as e:
            logger.error(f"Initial consistency check failed: {e}")
            self.stats['errors'] += 1
    
    async def _scheduler_loop(self):
        """定时任务调度循环"""
        while self.running:
            try:
                # 检查是否需要执行定时任务
                current_time = datetime.now()
                
                # 检查各种类型的定时任务
                await self._check_scheduled_tasks(current_time)
                
                # 等待下一个检查周期（每分钟检查一次调度）
                await asyncio.sleep(60)
                
            except asyncio.CancelledError:
                logger.info("Scheduler loop cancelled")
                break
            except Exception as e:
                logger.error(f"Error in scheduler loop: {e}")
                await asyncio.sleep(60)
    
    async def _check_scheduled_tasks(self, current_time: datetime):
        """检查计划任务"""
        schedule_config = self.config.get('scheduling', {})
        check_types = schedule_config.get('check_types', {})
        
        for check_type, check_config in check_types.items():
            if check_config.get('schedule') == 'manual':
                continue
            
            # 简化的调度逻辑（实际应该使用cron表达式解析）
            if await self._should_run_check(check_type, check_config, current_time):
                logger.info(f"Running scheduled check: {check_type}")
                try:
                    result = await self._run_check(check_type)
                    await self._process_check_result(result, check_type)
                except Exception as e:
                    logger.error(f"Scheduled check {check_type} failed: {e}")
                    self.stats['errors'] += 1
    
    async def _should_run_check(self, check_type: str, check_config: Dict, current_time: datetime) -> bool:
        """判断是否应该运行检查"""
        # 简化逻辑：基于检查间隔
        interval_hours = self.config['consistency_checks']['global_settings']['check_interval_hours']
        
        if not self.stats['last_check_time']:
            return True
        
        next_check_time = self.stats['last_check_time'] + timedelta(hours=interval_hours)
        return current_time >= next_check_time
    
    async def _run_check(self, check_type: str) -> Dict[str, Any]:
        """运行指定类型的检查"""
        schedule_config = self.config.get('scheduling', {})
        check_config = schedule_config.get('check_types', {}).get(check_type, {})
        
        # 设置检查参数
        duration_hours = check_config.get('duration_hours', 24)
        sample_size = check_config.get('sample_size', 1000)
        tables = check_config.get('tables', list(self.consistency_checker.table_mappings.keys()))
        
        # 设置时间范围
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=duration_hours)
        date_range = (start_time, end_time)
        
        # 执行检查
        reports = {}
        
        for table_name in tables:
            try:
                logger.info(f"Checking table {table_name} for {check_type}")
                report = await self.consistency_checker.check_table_consistency(
                    table_name, date_range, sample_size
                )
                reports[table_name] = report
                
                # 更新统计信息
                self.stats['total_inconsistencies_found'] += len(report.inconsistencies)
                
                critical_count = sum(
                    1 for inc in report.inconsistencies 
                    if inc.severity == SeverityLevel.CRITICAL
                )
                self.stats['critical_issues_found'] += critical_count
                
                # 记录一致性分数
                self.stats['last_consistency_scores'][table_name] = report.consistency_score
                
            except Exception as e:
                logger.error(f"Failed to check table {table_name}: {e}")
                self.stats['errors'] += 1
        
        return {
            'check_type': check_type,
            'timestamp': datetime.now().isoformat(),
            'reports': reports,
            'summary': self._generate_check_summary(reports)
        }
    
    def _generate_check_summary(self, reports: Dict[str, ConsistencyReport]) -> Dict[str, Any]:
        """生成检查摘要"""
        if not reports:
            return {}
        
        total_inconsistencies = sum(len(report.inconsistencies) for report in reports.values())
        total_records_checked = sum(report.total_records_checked for report in reports.values())
        
        avg_consistency_score = sum(report.consistency_score for report in reports.values()) / len(reports)
        
        severity_counts = {}
        for report in reports.values():
            for inc in report.inconsistencies:
                severity_counts[inc.severity.value] = severity_counts.get(inc.severity.value, 0) + 1
        
        return {
            'total_tables_checked': len(reports),
            'total_records_checked': total_records_checked,
            'total_inconsistencies': total_inconsistencies,
            'average_consistency_score': round(avg_consistency_score, 2),
            'severity_breakdown': severity_counts,
            'error_rate': total_inconsistencies / max(total_records_checked, 1)
        }
    
    async def _process_check_result(self, result: Dict[str, Any], check_type: str):
        """处理检查结果"""
        # 保存到历史记录
        self.check_history.append(result)
        if len(self.check_history) > self.max_history_size:
            self.check_history.pop(0)
        
        # 检查告警条件
        await self._check_alerts(result)
        
        # 生成报告
        await self._generate_report(result, check_type)
        
        # 更新统计信息
        self.stats['total_checks_performed'] += 1
        self.stats['last_check_time'] = datetime.now()
    
    async def _check_alerts(self, result: Dict[str, Any]):
        """检查告警条件"""
        alerting_config = self.config.get('alerting', {})
        thresholds = alerting_config.get('thresholds', {})
        
        summary = result.get('summary', {})
        
        # 检查一致性分数告警
        avg_score = summary.get('average_consistency_score', 100)
        if avg_score < thresholds.get('consistency_score', {}).get('critical', 85):
            await self._send_alert('critical', f"Consistency score critically low: {avg_score}%")
        elif avg_score < thresholds.get('consistency_score', {}).get('warning', 95):
            await self._send_alert('warning', f"Consistency score below warning threshold: {avg_score}%")
        
        # 检查错误率告警
        error_rate = summary.get('error_rate', 0)
        if error_rate > thresholds.get('error_rate', {}).get('critical', 0.05):
            await self._send_alert('critical', f"Error rate critically high: {error_rate:.2%}")
        elif error_rate > thresholds.get('error_rate', {}).get('warning', 0.01):
            await self._send_alert('warning', f"Error rate above warning threshold: {error_rate:.2%}")
    
    async def _send_alert(self, level: str, message: str):
        """发送告警"""
        logger.warning(f"ALERT [{level.upper()}]: {message}")
        
        # 这里可以实现实际的告警发送逻辑
        # 如邮件、Slack、webhook等
        alerting_config = self.config.get('alerting', {})
        
        # 示例：webhook告警
        webhook_config = alerting_config.get('notification_channels', {}).get('webhook', {})
        if webhook_config.get('enabled'):
            # 实现webhook发送逻辑
            pass
    
    async def _generate_report(self, result: Dict[str, Any], check_type: str):
        """生成检查报告"""
        reporting_config = self.config.get('reporting', {})
        
        if not reporting_config.get('generation', {}).get('auto_generate', True):
            return
        
        output_dir = reporting_config.get('generation', {}).get('output_directory', '/tmp')
        
        try:
            # 确保输出目录存在
            Path(output_dir).mkdir(parents=True, exist_ok=True)
            
            # 生成JSON报告
            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
            report_file = Path(output_dir) / f"consistency_report_{check_type}_{timestamp}.json"
            
            with open(report_file, 'w', encoding='utf-8') as f:
                json.dump(result, f, indent=2, default=str, ensure_ascii=False)
            
            logger.info(f"Report generated: {report_file}")
            
        except Exception as e:
            logger.error(f"Failed to generate report: {e}")
    
    async def stop_service(self):
        """停止一致性检查服务"""
        if not self.running:
            return
        
        logger.info("Stopping consistency service...")
        self.running = False
        
        if self.scheduler_task:
            self.scheduler_task.cancel()
            try:
                await self.scheduler_task
            except asyncio.CancelledError:
                pass
        
        if self.consistency_checker:
            await self.consistency_checker.cleanup()
        
        logger.info("Consistency service stopped")
    
    async def get_service_status(self) -> Dict[str, Any]:
        """获取服务状态"""
        status = {
            'running': self.running,
            'stats': self.stats.copy(),
            'config_loaded': self.config is not None,
            'checker_initialized': self.consistency_checker is not None,
            'system_info': {
                'cpu_percent': psutil.cpu_percent(),
                'memory_percent': psutil.virtual_memory().percent,
                'disk_percent': psutil.disk_usage('/').percent,
                'process_id': os.getpid()
            }
        }
        
        if self.stats['service_start_time']:
            uptime = datetime.now() - self.stats['service_start_time']
            status['uptime_seconds'] = uptime.total_seconds()
        
        return status
    
    async def run_manual_check(self, check_type: str = 'quick_check', 
                             table_name: str = None, 
                             hours_back: int = 24) -> Dict[str, Any]:
        """运行手动检查"""
        if not self.consistency_checker:
            raise RuntimeError("Consistency checker not initialized")
        
        try:
            if table_name:
                # 检查单个表
                end_time = datetime.now()
                start_time = end_time - timedelta(hours=hours_back)
                date_range = (start_time, end_time)
                
                report = await self.consistency_checker.check_table_consistency(
                    table_name, date_range
                )
                
                suggestions = await self.consistency_checker.generate_repair_suggestions(report)
                
                result = {
                    'check_type': 'manual_single_table',
                    'table_name': table_name,
                    'timestamp': datetime.now().isoformat(),
                    'report': asdict(report),
                    'suggestions': suggestions
                }
            else:
                # 运行预定义的检查类型
                result = await self._run_check(check_type)
            
            # 处理结果（但不保存到历史，因为是手动执行）
            await self._check_alerts(result)
            
            return result
            
        except Exception as e:
            logger.error(f"Manual check failed: {e}")
            raise


# 创建全局服务实例
consistency_service = ConsistencyService()

# 创建FastAPI应用
app = FastAPI(
    title="RedFire Data Consistency Service",
    description="数据一致性检查和监控服务",
    version="1.0.0"
)


@app.on_event("startup")
async def startup_event():
    """服务启动事件"""
    try:
        await consistency_service.initialize()
        logger.info("Consistency service API started")
    except Exception as e:
        logger.error(f"Failed to start consistency service: {e}")
        raise


@app.on_event("shutdown")
async def shutdown_event():
    """服务关闭事件"""
    await consistency_service.stop_service()
    logger.info("Consistency service API shutdown")


@app.get("/health")
async def health_check():
    """健康检查"""
    try:
        status = await consistency_service.get_service_status()
        return JSONResponse(
            status_code=200,
            content={
                "status": "healthy",
                "timestamp": datetime.now().isoformat(),
                "service": status
            }
        )
    except Exception as e:
        return JSONResponse(
            status_code=500,
            content={
                "status": "unhealthy",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }
        )


@app.get("/status")
async def get_status():
    """获取详细状态"""
    status = await consistency_service.get_service_status()
    return JSONResponse(content=status)


@app.get("/history")
async def get_check_history(limit: int = 10):
    """获取检查历史"""
    history = consistency_service.check_history[-limit:] if limit > 0 else consistency_service.check_history
    return JSONResponse(content=history)


@app.post("/service/start")
async def start_service(background_tasks: BackgroundTasks):
    """启动一致性检查服务"""
    try:
        background_tasks.add_task(consistency_service.start_service)
        return JSONResponse(content={"message": "Consistency service starting"})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/service/stop")
async def stop_service():
    """停止一致性检查服务"""
    try:
        await consistency_service.stop_service()
        return JSONResponse(content={"message": "Consistency service stopped"})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/check/run")
async def run_manual_check(
    check_type: str = 'quick_check',
    table_name: Optional[str] = None,
    hours_back: int = 24
):
    """运行手动检查"""
    try:
        result = await consistency_service.run_manual_check(check_type, table_name, hours_back)
        return JSONResponse(content=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/check/quick/{table_name}")
async def quick_table_check(table_name: str, hours_back: int = 6):
    """快速检查指定表"""
    try:
        result = await consistency_service.run_manual_check(
            'quick_check', table_name, hours_back
        )
        return JSONResponse(content=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/config")
async def get_config():
    """获取当前配置"""
    if not consistency_service.config:
        raise HTTPException(status_code=404, detail="Configuration not loaded")
    
    # 过滤敏感信息
    safe_config = consistency_service.config.copy()
    if 'databases' in safe_config:
        for db_type in ['mysql', 'postgresql']:
            if db_type in safe_config['databases']:
                safe_config['databases'][db_type]['password'] = "***"
    
    return JSONResponse(content=safe_config)


@app.get("/reports/latest")
async def get_latest_report():
    """获取最新的检查报告"""
    if not consistency_service.check_history:
        raise HTTPException(status_code=404, detail="No reports available")
    
    latest_report = consistency_service.check_history[-1]
    return JSONResponse(content=latest_report)


@app.get("/dashboard", response_class=HTMLResponse)
async def get_dashboard():
    """获取简单的Dashboard"""
    status = await consistency_service.get_service_status()
    
    html_content = f"""
    <!DOCTYPE html>
    <html>
    <head>
        <title>RedFire Data Consistency Dashboard</title>
        <style>
            body {{ font-family: Arial, sans-serif; margin: 20px; }}
            .status {{ padding: 10px; margin: 10px 0; border-radius: 5px; }}
            .healthy {{ background-color: #d4edda; color: #155724; }}
            .warning {{ background-color: #fff3cd; color: #856404; }}
            .error {{ background-color: #f8d7da; color: #721c24; }}
            .metric {{ display: inline-block; margin: 10px; padding: 10px; border: 1px solid #ddd; }}
        </style>
    </head>
    <body>
        <h1>RedFire Data Consistency Dashboard</h1>
        
        <div class="status {'healthy' if status['running'] else 'error'}">
            Service Status: {'Running' if status['running'] else 'Stopped'}
        </div>
        
        <h2>Statistics</h2>
        <div class="metric">
            <strong>Total Checks:</strong> {status['stats']['total_checks_performed']}
        </div>
        <div class="metric">
            <strong>Inconsistencies Found:</strong> {status['stats']['total_inconsistencies_found']}
        </div>
        <div class="metric">
            <strong>Critical Issues:</strong> {status['stats']['critical_issues_found']}
        </div>
        <div class="metric">
            <strong>Last Check:</strong> {status['stats']['last_check_time'] or 'Never'}
        </div>
        
        <h2>System Information</h2>
        <div class="metric">
            <strong>CPU:</strong> {status['system_info']['cpu_percent']:.1f}%
        </div>
        <div class="metric">
            <strong>Memory:</strong> {status['system_info']['memory_percent']:.1f}%
        </div>
        <div class="metric">
            <strong>Disk:</strong> {status['system_info']['disk_percent']:.1f}%
        </div>
        
        <h2>Recent Consistency Scores</h2>
        <ul>
    """
    
    for table, score in status['stats']['last_consistency_scores'].items():
        score_class = 'healthy' if score > 95 else 'warning' if score > 85 else 'error'
        html_content += f'<li class="{score_class}">{table}: {score:.1f}%</li>'
    
    html_content += """
        </ul>
        
        <p><a href="/docs">API Documentation</a></p>
    </body>
    </html>
    """
    
    return HTMLResponse(content=html_content)


def setup_signal_handlers():
    """设置信号处理器"""
    def signal_handler(signum, frame):
        logger.info(f"Received signal {signum}")
        asyncio.create_task(consistency_service.stop_service())
    
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)


async def run_service():
    """运行一致性检查服务"""
    setup_signal_handlers()
    
    try:
        await consistency_service.initialize()
        await consistency_service.start_service()
        
        # 保持服务运行
        while consistency_service.running:
            await asyncio.sleep(1)
            
    except KeyboardInterrupt:
        logger.info("Received keyboard interrupt")
    except Exception as e:
        logger.error(f"Service error: {e}")
    finally:
        await consistency_service.stop_service()


def run_api_server():
    """运行API服务器"""
    config = uvicorn.Config(
        app,
        host="0.0.0.0",
        port=8003,
        log_level="info"
    )
    server = uvicorn.Server(config)
    asyncio.run(server.serve())


if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="RedFire Data Consistency Service")
    parser.add_argument(
        "--mode", 
        choices=["service", "api", "check", "quick"], 
        default="service",
        help="运行模式: service(服务模式), api(API模式), check(单次检查), quick(快速检查)"
    )
    parser.add_argument(
        "--config", 
        help="配置文件路径",
        default=None
    )
    parser.add_argument(
        "--table",
        help="指定表名(仅在check模式下有效)",
        default=None
    )
    parser.add_argument(
        "--hours",
        type=int,
        help="检查时间范围(小时)",
        default=24
    )
    
    args = parser.parse_args()
    
    if args.config:
        consistency_service.config_path = args.config
    
    if args.mode == "service":
        # 服务模式：持续运行
        asyncio.run(run_service())
    elif args.mode == "api":
        # API模式：启动Web API服务
        run_api_server()
    elif args.mode == "check":
        # 检查模式：执行一次检查后退出
        async def run_check():
            await consistency_service.initialize()
            result = await consistency_service.run_manual_check(
                'manual_check', args.table, args.hours
            )
            print(json.dumps(result, indent=2, default=str))
            await consistency_service.stop_service()
        
        asyncio.run(run_check())
    elif args.mode == "quick":
        # 快速模式：快速检查后退出
        async def run_quick():
            mysql_config = {
                'host': 'localhost', 'port': 3306, 'user': 'redfire_user',
                'password': os.getenv('MYSQL_PASSWORD', 'password'),
                'db': 'redfire_business', 'charset': 'utf8mb4'
            }
            postgresql_config = {
                'host': 'localhost', 'port': 5432, 'user': 'redfire_analytics',
                'password': os.getenv('POSTGRESQL_PASSWORD', 'password'),
                'database': 'redfire_analytics'
            }
            
            result = await quick_consistency_check(
                mysql_config, postgresql_config, args.table, args.hours
            )
            print(json.dumps(result, indent=2, default=str))
        
        asyncio.run(run_quick())
